package com.yn.xgame;

import com.alibaba.fastjson.JSON;
import com.yn.xgame.message.C2GPbRedirectMessage;
import com.yn.xgame.message.Mq2ClientMessage;
import com.yn.xgame.message.MqBeanMessage;
import com.yn.xgame.pb.ClientGateProto;
import com.yn.xgame.util.Utils;
import io.netty.channel.ChannelHandlerContext;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * CommonMqMessageHandler
 * Created by wangboo on 17/7/13.
 */
public class CommonMqMessageHandler implements MQClient.Handler {

    private static final Logger logger = LoggerFactory.getLogger(CommonMqMessageHandler.class);

    private IServer msgProcessServer;
    private CtxManager ctxManager;

    public CommonMqMessageHandler(IServer msgProcessServer) {
        this.msgProcessServer = msgProcessServer;
    }

    public void setCtxManager(CtxManager ctxManager) {
        this.ctxManager = ctxManager;
    }

    @Override
    public void handleMessage(ConsumerRecord<String, byte[]> msg) {
//        logger.debug("recv MqMessage, topic: {}, userProp: {}, bin: {}", msg.topic(), msg.headers(),
//                msg.value());
        String tags = msg.key();
        if (tags == null) {
            logger.error("message tags not specified, msg: {}", msg);
            return;
        }
        switch (tags) {
            case MqTags.TO_CLIENT:
                receivePb(msg);
                return;
            case MqTags.C2G_REDIRECT:
                recveiveC2GRedirect(msg);
                return;
            case MqTags.BEAN:
                receiveBean(msg);
                return;
            default:
                logger.error("unhandled tags: {}", tags);
        }
    }

    private void recveiveC2GRedirect(ConsumerRecord<String, byte[]> msg) {
        try {
            long sid = -1;
            String fromGateTopic = null;
            for (Header header : msg.headers()) {
                if (header.key().equals("sid")) {
                    sid = Utils.getLong(header.value());
                }
                if (header.key().equals("topic")) {
                    fromGateTopic = new String(header.value(), Utils.charset);
                }
            }
            XAssert.nonNull(sid != -1, msg);
            XAssert.nonNull(fromGateTopic, msg);
            ClientGateProto.C2GMessage pb = ClientGateProto.C2GMessage.parseFrom(msg.value());
            if (logger.isDebugEnabled()) {
                logger.debug("recv pbRedirect: {}", pb);
            }
            C2GPbRedirectMessage pbMsg = new C2GPbRedirectMessage(sid, fromGateTopic, pb);
            msgProcessServer.pushMessage(pbMsg);
        } catch (Exception e) {
            logger.error("recveiveC2GRedirect error: ", e);
        }

    }

    private void receiveBean(ConsumerRecord<String, byte[]> msg) {
        String className = null, msgId = null, from = null;
        for (Header header : msg.headers()) {
            if (header.key().equals("class")) {
                className = new String(header.value(), Utils.charset);
            }
            if (header.key().equals("msgId")) {
                msgId = new String(header.value(), Utils.charset);
            }
            if (header.key().equals("from")) {
                from = new String(header.value(), Utils.charset);
            }
        }
        XAssert.nonNull(className, msg);
        XAssert.nonNull(msgId, msg);
        XAssert.nonNull(from, msg);
        XAssert.isTrue(msgId != null, msg);
        try {
            Object obj;
            if (MqTags.bean_type_map_str.equals(className)) {
                obj = JSON.parseObject(msg.value(), Map.class);
            } else {
                Class<?> clazz = Class.forName(className);
                obj = JSON.parseObject(msg.value(), clazz);
            }
            if (logger.isDebugEnabled()) {
                logger.debug("MQ Bean msg, from: {}, topic: {}, msgId: {}, body: {}", from, msg.topic(), msgId,
                        new String(msg.value()));
            }
            msgProcessServer.pushMessage(new MqBeanMessage(from, msg.topic(), msgId, obj));
        } catch (Exception e) {
            logger.error("error: ", e);
        }
    }

    private void receivePb(ConsumerRecord<String, byte[]> msg) {
        // 转发到客户端
        try {
//            String topic = msg.getToTopic();
            byte[] bin = msg.value();
            long sid = getSid(msg);
            // Map 在异步线程中get是安全的，不能有put操作
            ChannelHandlerContext ctx = ctxManager.get(sid);
            if (ctx != null) {
                Mq2ClientMessage mq2ClientMessage = new Mq2ClientMessage(sid, bin);
                ctx.writeAndFlush(mq2ClientMessage);
            } else {
                logger.error("can not sendInter msg to client, sid: {} not found, bin: {}", sid, bin);
            }
        } catch (Throwable e) {
            logger.error("handleReceivePb: ", e);
        }
    }

    private long getSid(ConsumerRecord<String, byte[]> msg) {
        for (Header header : msg.headers()) {
            if (header.key().equals("sid")) {
                return Utils.getLong(header.value());
            }
        }
        XAssert.isTrue(false, msg);
        return -1;
    }


}
